package com.xrui.hbase.conversion;


import com.tr.rts.cdf.CDFConsumer;
import com.util.ScalaUtil;
import com.xrui.hbase.Constants;
import com.xrui.hbase.HBaseCell;
import com.xrui.hbase.cdf.ByteConverter;
import com.xrui.hbase.model.RecordFields;
import org.apache.hadoop.hbase.util.Bytes;
import org.joda.time.DateTime;
import org.json4s.JsonAST;
import scala.Tuple2;

import java.util.Date;

/**
 * Converts an HBase cell encoded as a Protocol Buffer.
 */
public class CDFStoreBytesCellConverter implements CellConverter {
    /* The name of this provider */
    public static final String PROVIDER_NAME = "CDFStoreBytesCellConverter";
    public static final String KEY = "value";

    public static String currentAt = "";
    /** {@inheritDoc} */
    @Override
    public RecordFields convert(HBaseCell cell) {
        byte[] value = (byte[]) cell.getData();
        String columnName = cell.getColumn().toString();
        String changeId = cell.getEntityId().toString();
        RecordFields.Builder builder = RecordFields.Builder.newBuilder();
        builder.putField(Constants.ROW_KEY_FIELD_NAME, changeId);
        builder.putField(Constants.COLUMN_NAME_FIELD_NAME,columnName );
        builder.putField(Constants.TIMESTAMP_FIELD_NAME, cell.getTimestamp());

        if(columnName.equalsIgnoreCase("d:ENVELOPE"))
        {
             String  decodedvalue =ScalaUtil.decodeL1(value,changeId);
                    //CDFConsumer.retrieveL1FromAvro(value,changeId);
            builder.putField(KEY,  decodedvalue);

        }else if(columnName.contains("_SNAPSHOT")){
            byte[] snapBytes = ByteConverter.getRealBytes(value);
            JsonAST.JValue decodevalue = null;
//            if(currentAt.isEmpty()){
//                decodevalue  = CDFConsumer.retrieveRecordFromAvro(snapBytes,new Date(),changeId);
//            }
//            else{
//                decodevalue  = CDFConsumer.retrieveRecordFromAvro(snapBytes, DateTime.parse(currentAt).toDate(),changeId);
//            }
            byte[] snapBytes2 = ByteConverter.getRealBytes(value);
            ScalaUtil.decodeSnap(snapBytes);
            //JsonAST.JValue l2j = CDFConsumer.decodeSnapshot(snapBytes2);
            //= CDFConsumer.transform(l2j);
            String decodevalueStr = ScalaUtil.decodeSnap(snapBytes);
            builder.putField(KEY,decodevalueStr);
        }else {
            builder.putField(KEY, Bytes.toString(value));
        }


        return builder.build();
    }

    /** {@inheritDoc} */
    @Override
    public String getName() {
        return PROVIDER_NAME;
    }
    private void retrieveL1FromAvro(String  changeId ) {
        //Level1Decoder
    }
}
